package cn.itcast.function;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.IndexBean;
import cn.itcast.constant.Constant;
import cn.itcast.util.DateUtil;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;

import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;


public class IndexMinWindowFunction extends RichWindowFunction<CleanBean, IndexBean,String, TimeWindow> {
    MapState<String, IndexBean> mapState=null;
    @Override
    public void open(Configuration parameters) throws Exception {
        //2.初始化 MapState<String, IndexBean>
         mapState = getRuntimeContext().getMapState(new MapStateDescriptor<String, IndexBean>("mapState", String.class, IndexBean.class));
    }

    @Override
    public void apply(String s, TimeWindow window, Iterable<CleanBean> input, Collector<IndexBean> out) throws Exception {
    /**
     * 开发步骤：
     * 1.新建MinIndexWindowFunction 窗口函数
     * 2.初始化 MapState<String, IndexBean>
     * 3.记录最新指数
     * 4.获取分时成交额和成交数量
     * 5.格式化日期
     * 6.封装输出数据
     * 7.更新MapState
     */
    //     * 3.记录最新指数
        CleanBean line=null;
        for (CleanBean cleanBean : input) {
            if (line==null){
                line=cleanBean;
            }
            if (cleanBean.getEventTime()>line.getEventTime()){
                line=cleanBean;
            }
            //日成就总量
            Long tradeVolumn=0L;
            //日金额
            Long tradeAmt=0L;
            //* 4.获取分时成交额和成交数量
            IndexBean indexBean = mapState.get(line.getSecCode());
            if (indexBean!=null){
                tradeVolumn=line.getTradeVolumn()-indexBean.getTradeVolDay();
                tradeAmt=line.getTradeAmt()-indexBean.getTradeAmt();
            }
             // * 5.格式化日期
            Long aLong = DateUtil.longTimeTransfer(line.getEventTime(), Constant.format_yyyymmdd);
           // 6.封装输出数据
            IndexBean indexBean1 = new IndexBean(
                    line.getEventTime(),
                    line.getSecCode(),
                    line.getSecName(),
                    line.getPreClosePrice(),
                    line.getOpenPrice(),
                    line.getMaxPrice(),
                    line.getMinPrice(),
                    line.getTradePrice(),
                    tradeVolumn, tradeAmt,
                    line.getTradeVolumn(),
                    line.getTradeAmt(),
                    aLong,
                    line.getSource()
            );
            out.collect(indexBean1);
            mapState.put(line.getSecCode(),indexBean1);
        }

    }
}
